package com.bff.gaia.mix.api.xjoin2.operators;

import com.alibaba.fastjson.JSONObject;
import com.bff.gaia.api.java.tuple.Tuple5;
import com.bff.gaia.mix.api.operators.AbstractUdfMixStreamOperator;
import com.bff.gaia.mix.api.operators.TimestampedCollector;
import com.bff.gaia.mix.api.operators.TwoInputMixStreamOperator;
import com.bff.gaia.mix.api.xjoin2.client.mysql.all.MysqlAllXjoinInfo;
import com.bff.gaia.mix.api.xjoin2.core.enums.XjoinType;
import com.bff.gaia.mix.api.xjoin2.core.side.SideTableInfo;
import com.bff.gaia.mix.api.xjoin2.core.table.FieldInfo;
import com.bff.gaia.mix.api.xjoin2.core.table.JoinInfo;
import com.bff.gaia.mix.api.xjoin2.core.table.StreamTableInfo;
import com.bff.gaia.mix.api.xjoin2.core.table.XjoinInfo;
import com.bff.gaia.mix.api.xjoin2.functions.SyncXjoinFunction;
import com.bff.gaia.mix.runtime.streamrecord.MixStreamRecord;
import com.bff.gaia.types.Row;

import java.util.List;
import java.util.Map;
import java.util.Vector;
import java.util.concurrent.ConcurrentHashMap;

import static com.bff.gaia.mix.api.xjoin2.core.utils.KeyUtil.buildKey;

public class SyncXjoinOperator<IN1 extends Row, IN2 extends Tuple5<Integer, String, String, String, Integer>, OUT extends Row>
	extends AbstractUdfMixStreamOperator<OUT, SyncXjoinFunction>
	implements TwoInputMixStreamOperator<IN1, IN2, OUT> {

	private static final long serialVersionUID = 1L;

	protected XjoinInfo xjoinInfo;

	private transient TimestampedCollector<OUT> collector;

	private volatile boolean banner = true;
	private int version = -1;

	private Map<String, Row> streamTableCache;
	private Map<String, List<String>> sideTableCache;

	public SyncXjoinOperator(SyncXjoinFunction userFunction,
	                         StreamTableInfo streamTableInfo,
	                         SideTableInfo sideTableInfo,
	                         JoinInfo joinInfo,
	                         List<FieldInfo> outFieldInfoList) {
		super(userFunction);
		this.xjoinInfo = new MysqlAllXjoinInfo(streamTableInfo, sideTableInfo, joinInfo, outFieldInfoList);
	}

	@Override
	public void open() throws Exception {
		super.open();
		collector = new TimestampedCollector<OUT>(output);
		streamTableCache = new ConcurrentHashMap<>();//key
		sideTableCache = new ConcurrentHashMap<>();//key

		banner = true;// cache data
	}

	@Override
	public void close() throws Exception {
		super.close();
		sideTableCache.clear();
		streamTableCache.clear();
		System.gc();
	}

	@Override
	public void processElement1(MixStreamRecord<IN1> element) throws Exception {
		collector.setTimestamp(element);
		while (banner) {

		}
		String key = buildKey(element.getValue(), xjoinInfo.getStreamTableKeyFieldPos());
//		process data
		if (sideTableCache.containsKey(key)) {
			for (String jsonString : sideTableCache.get(key)) {
				userFunction.xjoin(element.getValue(), JSONObject.parseObject(jsonString), collector);
			}
		} else {
			if (xjoinInfo.getJoinInfo().getXjoinType().equals(XjoinType.LEFT)) {
				JSONObject nullObject = new JSONObject();
				for (String s : xjoinInfo.getSideSelectFields().split(",")) {
					nullObject.put(s, "");
				}
				userFunction.xjoin(element.getValue(), nullObject, collector);
			}
		}
//		String key = buildKey(element.getValue(), xjoinInfo.getStreamTableKeyFieldPos());
//		streamTableCache.put(key, element.getValue());
//		if (!banner) {
//			for (Map.Entry<String, Row> entry : streamTableCache.entrySet()) {
//				String cacheKey = entry.getKey();
//				//process data
//				if (sideTableCache.containsKey(cacheKey)) {
//					for (String jsonString : sideTableCache.get(cacheKey)) {
//						userFunction.xjoin(entry.getValue(), JSONObject.parseObject(jsonString), collector);
//					}
//					streamTableCache.remove(entry.getKey());
//				} else {
//					if (xjoinInfo.getJoinInfo().getXjoinType().equals(XjoinType.LEFT)) {
//						JSONObject nullObject = new JSONObject();
//						for (String s : xjoinInfo.getSideSelectFields().split(",")) {
//							nullObject.put(s, "");
//						}
//						userFunction.xjoin(entry.getValue(), nullObject, collector);
//					}
//				}
//			}
//		}
	}

	@Override
	public void processElement2(MixStreamRecord<IN2> element) throws Exception {
		collector.setTimestamp(element);
		int newVersion = element.getValue().getField(0);
		if (newVersion > this.version && !banner) {
			banner = true;
			if (!xjoinInfo.getJoinInfo().isIncLog()) {
				sideTableCache.clear();
			}
		}
		String action = element.getValue().getField(1);
		switch (action) {
			case "add":
			case "update": {

				String cacheKey = element.getValue().getField(2);
				List<String> tempMap = sideTableCache.containsKey(cacheKey) ?
					sideTableCache.get(cacheKey) :
					new Vector<>();
//				JSONObject value = element.getValue().getJSONObject("value");
				tempMap.add(element.getValue().getField(3));
				this.sideTableCache.put(cacheKey, tempMap);
				break;
			}
			case "delete_fullKey": {
				String cacheKey = element.getValue().getField(2);
				List<String> tempMap = sideTableCache.get(cacheKey);
				String fullKey = element.getValue().getField(3);
				tempMap.remove(fullKey);
				if (tempMap.isEmpty()) {
					this.sideTableCache.remove(cacheKey);
				} else {
					this.sideTableCache.put(cacheKey, tempMap);
				}

				break;
			}
			case "end": {
				this.version = newVersion;
				for (Map.Entry<String, Row> entry : streamTableCache.entrySet()) {
					String cacheKey = entry.getKey();
					//process data
					if (sideTableCache.containsKey(cacheKey)) {
						for (String jsonString : sideTableCache.get(cacheKey)) {
							userFunction.xjoin(entry.getValue(), JSONObject.parseObject(jsonString), collector);
						}
						streamTableCache.remove(entry.getKey());
					}
				}
				this.banner = false;
				break;
			}
			default: {
				//do nothing throw error
			}
		}


	}


}